package com.tinckay.common.config;

import com.tinckay.common.constant.AmqpConstant;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * @Author: Jean
 * @Desc:
 * @Date: 2018/9/3 14:23
 */
@Configuration
public class RabbitConfig {

    private Logger logger = LogManager.getLogger(RabbitConfig.class);

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 定制化amqp模版      可根据需要定制多个
     * <p>
     * <p>
     * 此处为模版类定义 Jackson消息转换器
     * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
     * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
     *
     * @return the amqp template
     */
    @Bean
    public AmqpTemplate amqpTemplate() {
        // rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        //rabbitTemplate.setMessageConverter(new FastJsonConverter());
        rabbitTemplate.setEncoding("UTF-8");
        /**
         * 消息确认，yml需要配置 publisher-confirms:true
         * 当一个消息被成功写入到RabbitMQ服务端时，就会自动的回调RabbitTemplate.ConfirmCallback接口内的confirm方法完成通知
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                String correlationId = null;
                if (null != correlationData) {
                    correlationId = correlationData.getId();
                }
                logger.info("消息发送到exchange成功,correlationId:{}", correlationId);
            } else {
                logger.debug("消息发送到exchange失败,原因: {}", cause);
            }
        });
        return rabbitTemplate;
    }

    /* ----------------------------------------------------------------------------Direct exchange binding--------------------------------------------------------------------------- */

    /**
     * 死信队列：
     * x-dead-letter-exchange: 用来设置死信后发送的交换机
     * x-dead-letter-routing-key：用来设置死信的routingKey
     * <p>
     * 死信队列设置:因为是队列到exchange的所以不能把交换机设置成internal,
     * 设置匹配的routingkey,表示消费失败的消息队列,最好可以一眼看出,是什么队列消费失败,所以我这里设置了queue_name.fail
     */

    //private final static Map<String, Object> DEAD_QUEUE_ARGS = new HashMap<String, Object>() {{
    //    put("x-dead-letter-exchange", AmqpConstant.DEAD_LETTER_EXCHANGE);
    //    put("x-dead-letter-routing-key", AmqpConstant.DEAD_LETTER_ROUTING_KEY);
    //}};
    //@Bean
    //public Exchange deadExchange() {
    //    return ExchangeBuilder
    //            .directExchange(AmqpConstant.DEAD_LETTER_EXCHANGE)
    //            .durable(true)
    //            .build();
    //}
    //@Bean
    //public Queue deadQueue() {
    //    return QueueBuilder
    //            .durable(AmqpConstant.DEAD_LETTER_QUEUE)
    //            .build();
    //}
    //@Bean
    //public Binding deadQueueBinding(Queue deadQueue, Exchange deadExchange) {
    //    return BindingBuilder
    //            .bind(deadQueue)
    //            .to(deadExchange)
    //            .with(AmqpConstant.DEAD_LETTER_ROUTING_KEY)
    //            .noargs();
    //}
    @Bean
    public FanoutExchange redisChangeFanoutExchange() {
        return new FanoutExchange(AmqpConstant.REDIS_CHANGE_FANOUT_EXCHANGE, true, false);
    }

    @Bean
    public Queue redisChangeQueue() {
        return QueueBuilder
                .durable(AmqpConstant.REDIS_CHANGE_QUEUE)
                .build();
    }

    @Bean
    public Binding redisChangeQueueBinding(Queue redisChangeQueue, FanoutExchange redisChangeFanoutExchange) {
        return BindingBuilder.bind(redisChangeQueue).to(redisChangeFanoutExchange);
    }

}
